/*
 * Copyright (c) 2018-2024 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "channel.h"
#include "common.h"
#include "subcmd/subcmd.h"
#include "quickpool.h"

#define OPEN_MPI 1
#define MPICH    2

char gstrJobAcctFile[PATH_MAX+1];
static struct conn_node *sptrTaskConn;
static unsigned int sintNumTask;
static unsigned int sintNumFinishTask;
static LIST_T *slistSignals;

extern char **environ;

static void handle_channels(void);
static void sig_handler(int);

int
main(int argc, char **argv)
{
    gf_set_ignorewarn(WARNING);
    if (gf_qp_init("qptask") < 0) {
        exit(-1);
    }
    gf_set_ignorewarn(NOLOG);

    int i;
    int fordev;
    if (getenv("DEVTASKDEBUG") != NULL) {

        fordev = atoi(getenv("DEVTASKDEBUG"));
        fordev = MIN(120, fordev);

        for (i=0; i<fordev;i++) {
            sleep(1);
        }

        for (i=0; i<argc; i++) {
            log_message(DEBUG, TASKSTART, "argv[%d]=%s", i, argv[i]);
        }
    }

    int rank=-1, mpiType=0;
    for (i=0; i<argc; i++) {

        if (0 == strcmp(argv[i], "--proxy-id")) {
            rank = atoi(argv[i+1]);
            mpiType = MPICH;
            break;
        } else if (0 == strcmp(argv[i], "ess_base_vpid") || 0 == strcmp(argv[i], "orte_ess_vpid")) {
            rank = atoi(argv[i+1]);
            mpiType = OPEN_MPI;
            break;
        }
    }

    char *strNode=NULL;
    for (i=0; i<argc; i++) {
        if (0 == strcmp(argv[i], "-x")) {
            strNode = copy_string(argv[i+1]);
            i += 2;
            break;
        }
    }

    char **envp = environ;
    int numEnv=0;
    while (NULL != envp[numEnv++]) {
        log_message(DEBUG, TASKSTART, "envrionment[%d]: %s", numEnv-1, envp[numEnv-1]);
    }

    int exitCode=0;

    JOBID_T jobId, jobIndex=0;

    char *strJobId;
    strJobId = getenv("QP_JOBID");
    if (NULL == strJobId) {
        log_message(ERR, NOCLASS, "%s: job task is not launched due to QP_JOBID not found.",
                    argv[0]);
        exit(97);
    }

    jobId = atoll(strJobId);

    char *strJobIndex;
    strJobIndex = getenv("QP_JOBINDEX");
    if (NULL != strJobIndex) {
        jobIndex = atoi(strJobIndex);
    }

    int numTask=0;
    char *strNumTask;
    strNumTask = getenv("QP_TASK_PERNODE");
    if (NULL != strNumTask) {
        numTask = atoi(strNumTask);
    }

    char *strExecUser;
    strExecUser = getenv("QPUSER");
    if (NULL == strExecUser) {
        strExecUser = getmy_uname();
    }

    char *strExecCwd;
    strExecCwd = getenv("QP_SUBCWD");
    if (NULL == strExecCwd) {
        char mycwd[PATH_MAX+1];
        strExecCwd = gf_getwd(mycwd);
    }

    char *strFirstNode;
    strFirstNode = getenv("QP_TASK_FIRSTNODE");
    if (NULL == strFirstNode) {
        strFirstNode = qp_getmyhostname();
    }

    char *strJobFile, *strJobFileDir;
    strJobFile = getenv("QP_JOBFILENAME");
    if (NULL != strJobFile) {

        strJobFileDir = strrchr(strJobFile, '/');
        if (NULL != strJobFileDir) {
            char *tmpDir = getenv("TMPDIR");
            if (NULL == tmpDir) {
                if (0 < rank) {
                    snprintf(gstrJobAcctFile, PATH_MAX, "%s/.%s.%d.acct",
                             gstrTmpDir, strJobFileDir + 1, rank);
                } else {
                    snprintf(gstrJobAcctFile, PATH_MAX, "%s/.%s.acct",
                             gstrTmpDir, strJobFileDir + 1);
                }
            } else {
                if (0 < rank) {
                    snprintf(gstrJobAcctFile, PATH_MAX, "%s/.%s.%d.acct",
                             tmpDir, strJobFileDir + 1, rank);
                } else {
                    snprintf(gstrJobAcctFile, PATH_MAX, "%s/.%s.acct",
                             tmpDir, strJobFileDir + 1);
                }
            }
        }
    }

    int port=0;
    char *strPort;
    strPort = getenv("QP_TASK_PORT");
    if (NULL != strPort) {
        port = atoi(strPort);
    }

    sintNumFinishTask = 0;

    if (NULL != strNode) {

        sintNumTask = 1;

        sptrTaskConn = (struct conn_node *)calloc(1, sizeof(struct conn_node));
        if (NULL == sptrTaskConn) {
            log_message(ERR, NOCLASS, "%s: job task is not launched due to calloc(%s) failure: %m.",
                        argv[0], sizeof(struct conn_node));
            exit(99);
        }

        sptrTaskConn->node_name = strNode;

        sptrTaskConn->task_argv = (char **)calloc(argc-i+2, sizeof(char *));
        if (NULL == sptrTaskConn->task_argv) {
            log_message(ERR, NOCLASS, "%s: job task is not launched due to calloc(%s) failure: %m.",
                        argv[0], (argc-i+1)*sizeof(char *));
            exit(99);
        }

        if (0 < jobIndex) {
            combine_jobid(&(sptrTaskConn->jobid), jobId, jobIndex);
        } else {
            sptrTaskConn->jobid = jobId;
        }

        sptrTaskConn->num_task = numTask;

        if (0 <= rank) {
            sptrTaskConn->node_rank = rank;
        }

        if (NULL != strFirstNode) {
            sptrTaskConn->firstnode_name = copy_string(strFirstNode);
        }
        sptrTaskConn->firstnode_port = port;

        if (NULL != strExecUser) {
            sptrTaskConn->execute_user = copy_string(strExecUser);
        }

        if (NULL != strExecCwd) {
            sptrTaskConn->cwd = copy_string(strExecCwd);
        }

        if (NULL != strJobFile) {
            sptrTaskConn->script = copy_string(strJobFile);
        }

        int j;
        if (OPEN_MPI == mpiType) {
            j = 0;
            sptrTaskConn->task_argv[j++] = copy_string("/bin/sh");
            sptrTaskConn->task_argv[j++] = copy_string("-c");

            char msg[2*PATH_MAX+1];
            memset(msg,0,sizeof(msg));

            for (; i<argc; i++) {
                char *sp = argv[i];

                if (0 == strcmp(sp, "-mca") && 0 == strcmp(argv[i+1], "plm_rsh_agent")) {
                    i += 2;
                    continue;
                } 

                if (0 == strcmp(sp, "-mca") && 0 == strcmp(argv[i+1], "plm_rsh_args")) {
                    i += 2;
                    continue;
                } 

                strncat(msg, " ", 2*PATH_MAX);
                strncat(msg, sp, 2*PATH_MAX);
            }

            sptrTaskConn->task_argv[j++] = copy_string(msg);
        } else {
            for (j=0; i<argc; i++) {
                char *sp = argv[i];
                if ('"' == *sp) {
                    sp ++;
                }

                if ('"' == sp[strlen(sp)-1]) {
                    sp[strlen(sp)-1] = '\0';
                }

                sptrTaskConn->task_argv[j++] = copy_string(sp);
            }
        }
        sptrTaskConn->task_argv[j] = NULL;

        execute_task(sptrTaskConn, g_jm_port, g_conntimeout, 0);
    } else { // start all job tasks by qptask

        Signal_(SIGINT, (SIGFUNCTYPE)sig_handler);
        Signal_(SIGHUP, (SIGFUNCTYPE)sig_handler);
        Signal_(SIGPIPE, (SIGFUNCTYPE)sig_handler);
        Signal_(SIGTTIN, (SIGFUNCTYPE)sig_handler);
        Signal_(SIGTTOU, (SIGFUNCTYPE)sig_handler);
        Signal_(SIGTSTP, (SIGFUNCTYPE)sig_handler);
        Signal_(SIGCONT, (SIGFUNCTYPE)sig_handler);

        slistSignals = gf_list_make("cached signal list");

        char *strNodes;
        strNodes = getenv("QP_HOSTS");
        if (NULL == strNodes) {
            log_message(ERR, NOCLASS, "%s: job task is not launched due to QP_HOSTS is not set.",
                        argv[0]);
            exit(96);
        }

        log_message(DEBUG, TASKSTART, "%s: QP_HOSTS=%s", argv[0], strNodes);

        struct nameList *nodeList;
        nodeList = longstring_2namelist(strNodes);
        if (NULL == nodeList) {
            log_message(ERR, NOCLASS, "%s: job task is not launched due to nodeList is NULL.",
                        argv[0]);
            exit(98);
        }

        sintNumTask = nodeList->num_name;

        sptrTaskConn = (struct conn_node *)calloc(nodeList->num_name, sizeof(struct conn_node));
        if (NULL == sptrTaskConn) {
            log_message(ERR, NOCLASS, "%s: job task is not launched due to calloc(%s) failure: %m.",
                        argv[0], nodeList->num_name*sizeof(struct conn_node));
            exit(99);
        }

        unsigned int i;
        for (i=0; i<nodeList->num_name; i++) {

            char *strNodeSlot;
            strNodeSlot = nodeList->names[i];

            char *token;
            unsigned int len = strlen(strNodeSlot);
            char *node = (char *)calloc(len+1, sizeof(char));
            if (NULL == node) {
                log_message(ERR, NOCLASS, "%s: job task is not launched due to calloc(%s) failure: %m.",
                            argv[0], len+1);
                exit(99);
            }

            token = get_string_token(&strNodeSlot, ":", node, len+1);
            if (NULL == token) {
                log_message(ERR, NOCLASS, "%s: job task is not launched due to node name is not found around <%s>.",
                            argv[0], nodeList->names[i]);
                exitCode = 95;
                break;
            }
            sptrTaskConn[i].node_name = copy_string(token);

            len = strlen(strNodeSlot);
            token = get_string_token(&strNodeSlot, ":", node, len+1);
            if (NULL == token) {
                log_message(ERR, NOCLASS, "%s: job task is not launched due to task number is not found around <%s>.",
                            argv[0], nodeList->names[i]);
                exitCode = 94;
                break;
            }

            sptrTaskConn[i].num_task = gf_atoi(token, INFINIT_INT, 0);
            if (INFINIT_INT == sptrTaskConn[i].num_task) {
                if (0 != numTask) {
                    sptrTaskConn[i].num_task = numTask;
                } else {
                    log_message(ERR, NOCLASS, "%s: job task is not launched due to task number <%s> is not an integer between %d-%d.",
                                argv[0], token, 1, INFINIT_INT);
                    exitCode = 93;
                    break;
                }
            }
            FREEUP(node);

            sptrTaskConn[i].node_rank = i;
            if (NULL != strFirstNode) {
                sptrTaskConn[i].firstnode_name = copy_string(strFirstNode);
            }
            sptrTaskConn[i].firstnode_port = port;

            if (NULL != strExecUser) {
                sptrTaskConn[i].execute_user = copy_string(strExecUser);
            }

            if (NULL != strExecCwd) {
                sptrTaskConn[i].cwd = copy_string(strExecCwd);
            }

            if (NULL != strJobFile) {
                sptrTaskConn[i].script = copy_string(strJobFile);
            }

            sptrTaskConn[i].task_argv = (char **)calloc(argc, sizeof(char *));
            if (NULL == sptrTaskConn[i].task_argv) {
                log_message(ERR, NOCLASS, "%s: job task is not launched due to calloc(%s) failure: %m.",
                            argv[0], argc*sizeof(char *));
                exitCode = 99;
                break;
            }

            int j, k=0;
            if (OPEN_MPI == mpiType) {

                sptrTaskConn[i].task_argv[k++] = copy_string("bash");
                sptrTaskConn[i].task_argv[k++] = copy_string("-c");

                char msg[2*PATH_MAX+1];
                memset(msg,0,sizeof(msg));

                for (j=1; j<argc; j++) {
                    char *sp = argv[j];

                    if (0 == strcmp(sp, "-mca") && 0 == strcmp(argv[i+1], "plm_rsh_agent")) {
                        j += 2;
                        continue;
                    } 

                    if (0 == strcmp(sp, "-mca") && 0 == strcmp(argv[i+1], "plm_rsh_args")) {
                        j += 2;
                        continue;
                    } 

                    strncat(msg, " ", 2*PATH_MAX);
                    strncat(msg, sp, 2*PATH_MAX);
                }

                sptrTaskConn[i].task_argv[k++] = copy_string(msg);
            } else {
                for (j=1; j<argc; j++) {

                    char *sp = argv[j];
                    if ('"' == *sp) {
                        sp ++;
                    }

                    if ('"' == sp[strlen(sp)-1]) {
                        sp[strlen(sp)-1] = '\0';
                    }

                    sptrTaskConn[i].task_argv[k++] = copy_string(sp);
                }
            }

            sptrTaskConn[i].task_argv[k] = NULL;

            if (0 < jobIndex) {
                combine_jobid(&(sptrTaskConn[i].jobid), jobId, jobIndex);
            } else {
                sptrTaskConn[i].jobid = jobId;
            }

            execute_task(&(sptrTaskConn[i]), g_jm_port, g_conntimeout, 0);
        }

        free_namelist(nodeList);
    }

    for (;;) {

        gf_handle_socket(5000);
        handle_channels();

        if (sintNumFinishTask >= sintNumTask) {
            break;
        }

        sigset_t newMask;
        sigset_t oldMask;
        gf_block_allsignals(&newMask, &oldMask);

        LIST_PROXY_T *pxy;
        while (NULL != (pxy=(LIST_PROXY_T *)gf_list_pop(slistSignals))) {

            int *signalVal;
            signalVal = (int *)(pxy->subject);

            unsigned int k;
            for (k=0; k<sintNumTask; k++) {
                send_signal_2task(&sptrTaskConn[k], *signalVal);
            }

            FREEUP(signalVal);
            FREEUP(pxy);
        }

        gf_unblock_signals(&oldMask);
    }

    FREEUP(sptrTaskConn);

    exit(exitCode);
}

/**
 * @brief       Write job task usage into job task acct file.
 *
 * @param[in]   numtask      #1: number of total task of the job
 * @param[in]   tasks        #2: tasks of the job
 *
 * @note        Only log acct for the job who runs on multiple nodes.
 *              JM on first node would read this file to collect job usage
 *              on different nodes.
 */
void
log_job_acct(int numtask, struct conn_node *tasks)
{
    FILE *fileAcct;
    fileAcct = fopen(gstrJobAcctFile, "w");
    if (NULL == fileAcct) {
        log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "fopen", gstrJobAcctFile);
        return;
    }

    struct record taskRecord;
    snprintf(taskRecord.version, MAXVERSIONLEN, "%d", QP_XDR_VERSION);
    taskRecord.type = RECORD_TYPE_JOB_ACCT;
    taskRecord.log_time = time(NULL);

    unsigned int num = 2; // gintExecJobId, numtask

    num += 7*numtask; // node_rank, pid, pgid, wait, exit_code, start_time, end_time

    int i;
    for (i=0; i<numtask; i++) {

        if (NULL == tasks[i].inst_usage) {
            continue;
        }

        num += 6; //mem, swap, utime, stime, npids, npgids

        num += tasks[i].inst_usage->npids; //pid, ppid, pgid
        num += tasks[i].inst_usage->npgids; //pgid
    }

    taskRecord.KVs = (key_value_t *)calloc(num, sizeof(key_value_t));
    if (NULL == taskRecord.KVs) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", num*sizeof(key_value_t));
        return;
    }

    num = 0;
    taskRecord.KVs[num].key = RECORD_KEY_JOBID;
    taskRecord.KVs[num].value = (char *)calloc(32, sizeof(char));
    if (NULL == taskRecord.KVs[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 32);
        return;
    }
    snprintf(taskRecord.KVs[0].value, 32, "%lld", tasks->jobid);
    num ++;

    taskRecord.KVs[num].key = RECORD_KEY_NUM_TASK;
    taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
    if (NULL == taskRecord.KVs[num].value) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
        return;
    }
    snprintf(taskRecord.KVs[num].value, 12, "%d", numtask);
    num ++;

    for (i=0; i<numtask; i++) {

        taskRecord.KVs[num].key = RECORD_KEY_NODE_RANK;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%d", tasks[i].node_rank);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_JOB_PID;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%d", tasks[i].pid);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_JOB_PGID;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%d", tasks[i].pgid);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_TASK_STATUS;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%d", tasks[i].wait);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_EXIT_CODE;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%d", tasks[i].exit_code);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_START_TIME;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%ld", tasks[i].start_time);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_END_TIME;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%ld", tasks[i].end_time);
        num ++;

        if (NULL == tasks[i].inst_usage) {
            continue;
        }

        taskRecord.KVs[num].key = RECORD_KEY_USAGE_MEM;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%ld", tasks[i].inst_usage->mem);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_USAGE_SWAP;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%ld", tasks[i].inst_usage->swap);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_USAGE_UTIME;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%ld", tasks[i].inst_usage->utime);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_USAGE_STIME;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%ld", tasks[i].inst_usage->stime);
        num ++;

        taskRecord.KVs[num].key = RECORD_KEY_USAGE_NPID;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%d", tasks[i].inst_usage->npids);
        num ++;

        unsigned int j;
        for (j=0; j<tasks[i].inst_usage->npids; j++) {
            taskRecord.KVs[num].key = RECORD_KEY_USAGE_PIDINFO;
            taskRecord.KVs[num].value = (char *)calloc(36, sizeof(char));
            if (NULL == taskRecord.KVs[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 36);
                return;
            }

            snprintf(taskRecord.KVs[num].value, 36, "%d %d %d",
                     tasks[i].inst_usage->pidinfo[j].pid,
                     tasks[i].inst_usage->pidinfo[j].ppid,
                     tasks[i].inst_usage->pidinfo[j].pgid);
            num ++;
        }

        taskRecord.KVs[num].key = RECORD_KEY_USAGE_NPGID;
        taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
        if (NULL == taskRecord.KVs[num].value) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
            return;
        }
        snprintf(taskRecord.KVs[num].value, 12, "%d", tasks[i].inst_usage->npgids);
        num ++;

        for (j=0; j<tasks[i].inst_usage->npgids; j++) {
            taskRecord.KVs[num].key = RECORD_KEY_USAGE_PGID;
            taskRecord.KVs[num].value = (char *)calloc(12, sizeof(char));
            if (NULL == taskRecord.KVs[num].value) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, 12);
                return;
            }
            snprintf(taskRecord.KVs[num].value, 12, "%d", tasks[i].inst_usage->pgid[i]);
            num ++;
        }
    }

    taskRecord.numKVs = num;

    if (write_record(fileAcct, &taskRecord) == -1) {
        fclose(fileAcct);
        return;
    }

    fclose(fileAcct);

    return;
} // end function log_job_acct

/* ------------------------------------------------
 *
 *  below are static functions used in this file
 *
 * ------------------------------------------------ */

/**
 * @brief socket handler: BATCH_STATUS_TASK request.
 *
 * @param[in]   fd           #1: socket fd
 * @param[in]   hdr          #2: XDR header
 * @param[in]   xdrs         #3: xdrs for decode new job request
 *
 * @note        handle task usage from remote execution node and log it into
 *              job acct file.
 */
static void
handle_task_status(int fd, struct header *hdr, XDR *xdrs)
{
    struct request req;
    memset(&req, 0, sizeof(struct request));

    if (!gf_xdr_request(xdrs, &req, hdr, NULL)) {
        log_message(ERR, NOCLASS, "%s: gf_xdr_request failed.", __func__);
        return;
    }

    int rank=-1;
    unsigned int i;
    for (i=0; i<req.number; i++) {
        switch (req.keyvalues[i].key) {
        case REQUEST_KEY_NODE_RANK:
            rank = atoi(req.keyvalues[i].value);
            break;
        }

        if (0 <= rank) {
            break;
        }
    }

    if (0 > rank) {
        log_message(ERR, NOCLASS, "%s: node rank is not found.", __func__);
        return;
    }

    for (i=0; i<sintNumTask; i++) {

        if (sptrTaskConn[i].conn_fd == fd) {
            break;
        }
    }

    if (i >= sintNumTask) {
        log_message(ERR, NOCLASS, "%s: not found conn_fd <%d> in my task list.",
                    __func__, fd, sintNumTask);
        return;
    }

    struct conn_node *task;
    task = &(sptrTaskConn[i]);
    if (NULL == task->inst_usage) {

        task->inst_usage = (struct jobResourceUsage *)calloc(1, sizeof(struct jobResourceUsage));
        if (NULL == task->inst_usage) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                        sizeof(struct jobResourceUsage));
            return;
        }
    }

    struct jobResourceUsage *taskUsage;
    taskUsage = task->inst_usage;

    if (0 < taskUsage->npids || 0 < taskUsage->npgids) {
        gf_xdr_free(xdr_job_usage, taskUsage);
    }
    taskUsage->utime = 0;
    taskUsage->stime = 0;

    unsigned int numPid=0, numPGid=0;
    for (i=0; i<req.number; i++) {
        switch (req.keyvalues[i].key) {
        case REQUEST_KEY_JOBPID:
            task->pid = atoi(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_JOBPGID:
            task->pgid = atoi(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_EXITSTATUS:
            task->wait = atoi(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_START_TIME:
            task->start_time = atol(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_USAGE_MEM:
            taskUsage->mem = atol(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_USAGE_SWAP:
            taskUsage->swap = atol(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_USAGE_UTIME:
            taskUsage->utime = atol(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_USAGE_STIME:
            taskUsage->stime = atol(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_USAGE_NPID:
            taskUsage->npids = atoi(req.keyvalues[i].value);
            if (0 < taskUsage->npids) {
                taskUsage->pidinfo = (struct pidInfo *)calloc(taskUsage->npids,
                                                              sizeof(struct pidInfo));
                if (NULL == taskUsage->pidinfo) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                                taskUsage->npids*sizeof(struct pidInfo));
                    return;
                }
            }
            break;
        case REQUEST_KEY_USAGE_PIDINFO:
            sscanf(req.keyvalues[i].value, "%d %d %d", &(taskUsage->pidinfo[numPid].pid),
                   &(taskUsage->pidinfo[numPid].ppid), &(taskUsage->pidinfo[numPid].pgid));
            numPid++;
            break;
        case REQUEST_KEY_USAGE_NPGID:
            taskUsage->npgids = atoi(req.keyvalues[i].value);
            if (0 < taskUsage->npgids) {
                taskUsage->pgid = (int *)calloc(taskUsage->npgids, sizeof(int));
                if (NULL == taskUsage->pgid) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                                taskUsage->npgids*sizeof(int));
                    return;
                }
            }
            break;
        case REQUEST_KEY_USAGE_PGID:
            taskUsage->pgid[numPGid++] = atol(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_END_TIME:
            task->end_time = atol(req.keyvalues[i].value);
            break;
        case REQUEST_KEY_FAIL_REASON:
            task->exit_code = atoi(req.keyvalues[i].value);
            break;
        }
    }

    log_job_acct(sintNumTask, sptrTaskConn);

    if (0 < task->end_time) {
        gf_chan_close(task->conn_fd);
        sintNumFinishTask ++;
    }

    return;
} // end function handle_task_status

/**
 * @brief       Handle channel.
 *
 * @note        Get socket fds which have data and handle them one by one.
 */
static void
handle_channels(void)
{          
    int *handleFds=NULL;
    unsigned int numHandles = gf_get_fdlist(&handleFds);

    if (0 == numHandles) {
        FREEUP(handleFds);
        return;
    }

    LIST_T **dataList;
    dataList = (LIST_T **)calloc(numHandles, sizeof(LIST_T *));
    if (NULL == dataList) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", numHandles*sizeof(LIST_T *));
        FREEUP(handleFds);
        return;
    }

    unsigned int i;
    for (i=0; i<numHandles; i++) {

        int fd = handleFds[i];
        int numList = gf_get_data(handleFds[i], &(dataList[i]));

        if (0 >= numList) {
            continue;
        }

        socklen_t len;
        struct sockaddr_in from;
        len = sizeof(struct sockaddr_in);
        if (getpeername(fd, (struct sockaddr *)&from, &len) < 0) {
            log_message(ERR, NOCLASS, "%s: getpeername() failed on %d. %m", __func__, fd);

            gf_chan_close(fd);
            handleFds[i] = -1;
            continue;
        }

        char addrbuf[24+1];
        gf_sockaddr_str(&from, addrbuf);

        struct chan_data *recvData;
        struct chan_data *nextData;
        for (recvData = (struct chan_data *)dataList[i]->forw;
             recvData != (struct chan_data *)dataList[i]; recvData = nextData) {

            nextData = recvData->forw;

            if (NULL == recvData) {
                continue;
            }

            struct header *hdr = &(recvData->hdr);
            log_message(DEBUG, CHANNEL, "%s: got request[%d] on fd[%d] from <%s>.",
                        __func__, hdr->reqcode, handleFds[i], addrbuf);

            XDR xdrs;
            xdrmem_create(&xdrs, recvData->data, recvData->datalen, XDR_DECODE);

            struct auth auth;
            memset(&auth, 0, sizeof(struct auth));

            ee_request_t reqCode;
            reqCode = (ee_request_t)(hdr->reqcode);

            switch (reqCode) {
            case BATCH_STATUS_TASK:
                handle_task_status(fd, hdr, &xdrs);
                break;
            default:
                break;
            }

            gf_xdr_free(gf_xdr_auth, &auth);
            xdr_destroy(&xdrs);
            gf_clean_recvdata(fd, recvData);
        }

    }

    FREEUP(dataList);
    FREEUP(handleFds);

    return;
} // end function handle_channels

static void
sig_handler(int signum)
{
    LIST_PROXY_T *pxy;
    int *sig;

    sig = (int *)calloc(1, sizeof(int));
    if (NULL == sig) {
        return;
    }

    *sig = signum;

    pxy = (LIST_PROXY_T *)calloc(1, sizeof(struct list_proxy));
    if (NULL == pxy) {
        return;
    }

    pxy->subject = (void *)sig;
    gf_list_insert(slistSignals, slistSignals->forw, (LIST_T *)pxy);
    
    return;
}
